#include "ul_pipe_task_queue.h"
#include <unistd.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/socket.h>
#include "ul_define.h"
#include "ul_log_adapter.h"
#include "ul_util.h"
using namespace LinuxNetworkUtility;
namespace 
{
  void close_pipe(int *fd)
  {
    if (NULL != fd)
    {
      close(fd[0]);
      close(fd[1]);
      fd[0] = -1;
      fd[1] = -1;
    }
  }

  int write_all(int fd, char *data, int32_t len)
  {
    int err = 0;
    int32_t writed_len = 0;
    int32_t cur_writed_len = 0;
    while (writed_len < len)
    {
      cur_writed_len = write(fd, data + writed_len, len - writed_len);
      if (cur_writed_len < 0)
      {
        if (EAGAIN == errno || EINTR == errno)
        {
        }
        else
        {
          UL_LOG_WARNING("fail to write to fd [errno:%d]", errno);
          err = ERRNO2ERR(errno);
          break;
        }
      }
      else
      {
        writed_len += cur_writed_len;
      }
    }
    return err;
  }

  int fast_read(const int fd, void *buf, const int64_t size, const int64_t timeoutus)
  {
    int err = 0;
    int64_t readed_len = 0;
    bool readable = false;
    readed_len = read(fd,  buf, size);
    if (readed_len <= 0)
    {
      if (0 == readed_len || EAGAIN ==  errno || EINTR == errno)
      {
        readed_len = 0;
        readable = ul_check_fd_readable(fd, timeoutus);
      }
      else
      {
        err = ERRNO2ERR(errno);
      }
    }
    else
    {
      err = readed_len;
    }
    if (0 == err && readed_len == 0 && readable)
    {
      readed_len = read(fd,  buf, size);
      err = readed_len;
      if (0 == readed_len)
      {
        errno = EAGAIN;
      }
    }
    return err;
  }
}

LinuxNetworkUtility::UlPipeMultiProducerSingleConsumerTQ::UlPipeMultiProducerSingleConsumerTQ()
{
  fd_pair_.fd_pair_.fd_[0] = -1;
  fd_pair_.fd_pair_.fd_[1] = -1;
}


void LinuxNetworkUtility::UlPipeMultiProducerSingleConsumerTQ::clear()
{
  close_pipe(fd_pair_.fd_pair_.fd_);
}

LinuxNetworkUtility::UlPipeMultiProducerSingleConsumerTQ::~UlPipeMultiProducerSingleConsumerTQ()
{
  clear();
}

int LinuxNetworkUtility::UlPipeMultiProducerSingleConsumerTQ::init(const int64_t producer_num)
{
  int err = 0;
  UNUSED(producer_num);
  if (pipe(fd_pair_.fd_pair_.fd_) != 0)
  /// if (socketpair(AF_UNIX, SOCK_STREAM,0, fd_pair_.fd_pair_.fd_))
  {
    err = ERRNO2ERR(errno);
    UL_LOG_WARNING("open pipe failed [errno:%d]",errno);
  }
  else
  {
    int flags = 0;
    flags = fcntl(fd_pair_.pipe_fds_.read_fd_, F_GETFL, 0);
    flags |= O_NONBLOCK;
    fcntl(fd_pair_.pipe_fds_.read_fd_, F_SETFL, flags);
  }
  return err;
}

int LinuxNetworkUtility::UlPipeMultiProducerSingleConsumerTQ::push(const void *tsk, 
                                                                   const int64_t producer_id)
{
  UNUSED(producer_id);
  int err = 0;
  if (0 > fd_pair_.fd_pair_.fd_[0] || 0 > fd_pair_.fd_pair_.fd_[0])
  {
    errno = EPIPE;
    err = ERRNO2ERR(errno);
    UL_LOG_WARNING("pipe broken");
  }
  else
  {
    if (write(fd_pair_.pipe_fds_.write_fd_, &tsk, sizeof(void*)) != sizeof(void*))
    {
      UL_LOG_WARNING("write to pipe fail [errno:%d]", errno);
      err = ERRNO2ERR(errno);
      clear();
    }
  }
  return err;
}

int LinuxNetworkUtility::UlPipeMultiProducerSingleConsumerTQ::pop(void **tsk, const int64_t ntsk, 
                                                                  int64_t timeout_us)
{
  int err = 0;
  int32_t readed_len = 0;
  if (NULL == tsk || ntsk <= 0)
  {
    errno  = EINVAL;
    err = ERRNO2ERR(errno);
    UL_LOG_WARNING("argument error [tsk:%p,ntsk:%lld]", tsk, ntsk);
  }
  if (0 == err && ( 0 > fd_pair_.fd_pair_.fd_[0] || 0 > fd_pair_.fd_pair_.fd_[0]))
  {
    errno = EPIPE;
    err = ERRNO2ERR(errno);
    UL_LOG_WARNING("pipe broken");
  }

  if (0 == err)
  {
    readed_len = fast_read(fd_pair_.pipe_fds_.read_fd_, tsk, sizeof(void*)*ntsk, timeout_us);
    if (readed_len > 0 && readed_len % sizeof(void*) == 0)
    {
      /// do nothing
    }
    else
    {
      if (EAGAIN == errno || EINTR == errno)
      {
        err = ERRNO2ERR(errno);
      }
      else
      {
        UL_LOG_WARNING("read from pipe error [errno:%d]", errno);
        err = ERRNO2ERR(errno);
        clear();
      }
    }
  }
  if (0 == err && readed_len > 0)
  {
    err = readed_len/sizeof(void*);
  }
  return err;
}


LinuxNetworkUtility::UlPipeQueenWorkerTQ::UlPipeQueenWorkerTQ()
{
  consumer_num_ = 0;
  fd_pairs_ = NULL;
  queen_poll_fds_ = NULL;
  next_push_pipe_idx_ = 0;
}

void LinuxNetworkUtility::UlPipeQueenWorkerTQ::clear()
{
  if (NULL != fd_pairs_)
  {
    for (int32_t i = 0; i < consumer_num_; i++)
    {
      close_pipe(fd_pairs_[i].fd_pair_.fd_);
    }
    delete [] fd_pairs_;
    fd_pairs_ = NULL;
  }
  if (NULL != queen_poll_fds_)
  {
    delete [] queen_poll_fds_;
    queen_poll_fds_ = NULL;
  }
  consumer_num_ = 0;
  next_push_pipe_idx_ = 0;
}

LinuxNetworkUtility::UlPipeQueenWorkerTQ::~UlPipeQueenWorkerTQ()
{
  clear();
}

int LinuxNetworkUtility::UlPipeQueenWorkerTQ::init(const int64_t worker_bee_num)
{
  int err = 0;
  if (NULL != fd_pairs_)
  {
    UL_LOG_WARNING("task queue was already initialized");
    errno = EALREADY;
    err =  ERRNO2ERR(errno);
  }
  if (worker_bee_num <= 0)
  {
    UL_LOG_WARNING("invalid argument [worker_bee_num:%lld]", worker_bee_num);
    errno = EINVAL;
    err = ERRNO2ERR(errno);
  }

  if (0 == err)
  {
    consumer_num_ = worker_bee_num;
    fd_pairs_ = new(std::nothrow) PipeFdPairs[worker_bee_num];
    if (NULL == fd_pairs_)
    {
      UL_LOG_WARNING("fail to allocate memory for fd pair array [errno:%d]", errno);
      err = ERRNO2ERR(errno);
    }
    else
    {
      queen_poll_fds_ = new(std::nothrow) pollfd[worker_bee_num];
      if (NULL == queen_poll_fds_)
      {
        UL_LOG_WARNING("fail to allocate memory for pollfd array [errno:%d]", errno);
        err = ERRNO2ERR(errno);
      }
    }
    if (0 == err)
    {
      for (int32_t i = 0; i < consumer_num_; i++)
      {
        fd_pairs_[i].fd_pair_.fd_[0] = -1;
        fd_pairs_[i].fd_pair_.fd_[1] = -1;
      }
      for (int32_t i = 0; i < consumer_num_; i++)
      {
        /// if (pipe(fd_pairs_[i].fd_pair_.fd_) != 0)
        if (socketpair(AF_UNIX, SOCK_STREAM,0, fd_pairs_[i].fd_pair_.fd_) != 0)
        {
          err = ERRNO2ERR(errno);
          UL_LOG_WARNING("fail to open pipe [errno:%d]",errno);
          break;
        }
        else
        {
          int flags = 0;
          flags = fcntl(fd_pairs_[i].pipe_fds_.queen_fd_, F_GETFL, 0);
          flags |= O_NONBLOCK ;
          fcntl(fd_pairs_[i].pipe_fds_.queen_fd_, F_SETFL, flags);

          flags = fcntl(fd_pairs_[i].pipe_fds_.worker_fd_, F_GETFL, 0);
          flags |= O_NONBLOCK;
          fcntl(fd_pairs_[i].pipe_fds_.worker_fd_, F_SETFL, flags);

          queen_poll_fds_[i].fd = fd_pairs_[i].pipe_fds_.queen_fd_;
          queen_poll_fds_[i].events = POLLIN;
        }
      }
    }
  }
  if ((NULL != fd_pairs_ || NULL != queen_poll_fds_) && 0 != err)
  {
    clear();
  }
  return err;
}

int LinuxNetworkUtility::UlPipeQueenWorkerTQ::dispatch_task(void *tsk)
{
  int err = 0;
  if (NULL == fd_pairs_)
  {
    errno = EINVAL;
    err = ERRNO2ERR(errno);
    UL_LOG_WARNING("task queue not initialized yet, initialize it first");
  }
  int64_t cur_pipe_idx = next_push_pipe_idx_;
  if (0 == err)
  {
    /// find first idle consumer
    int64_t checked_fd_num = 0;
    for (checked_fd_num = 0; checked_fd_num < consumer_num_; checked_fd_num++)
    {
      if (!ul_check_fd_readable(fd_pairs_[cur_pipe_idx].pipe_fds_.worker_fd_))
      {
        /// if (write(fd_pairs_[cur_pipe_idx].pipe_fds_.queen_fd_,&tsk, sizeof(void*)) != sizeof(void*))
        if (write_all(fd_pairs_[cur_pipe_idx].pipe_fds_.queen_fd_,(char*)&tsk, sizeof(void*)) != 0)
        {
          UL_LOG_ERROR("fail to write data to pipe [idx:%lld,errno:%d]", cur_pipe_idx, errno);
          queen_poll_fds_[cur_pipe_idx].fd = -1;
          close_pipe(fd_pairs_[cur_pipe_idx].fd_pair_.fd_);
        }
        else
        {
          next_push_pipe_idx_ = (cur_pipe_idx + 1)%consumer_num_;
          break;
        }
      }
      else
      {
        cur_pipe_idx ++;
        cur_pipe_idx = cur_pipe_idx%consumer_num_;
      }
    }
    /// there is no idle consumer, find first correct consumer
    if (checked_fd_num >= consumer_num_)
    {
      cur_pipe_idx = next_push_pipe_idx_;
      for (checked_fd_num = 0; checked_fd_num < consumer_num_; checked_fd_num++)
      {
        if (fd_pairs_[cur_pipe_idx].pipe_fds_.worker_fd_ >= 0)
        {
          ///if (write(fd_pairs_[cur_pipe_idx].pipe_fds_.queen_fd_,&tsk, sizeof(void*)) != sizeof(void*))
          if (write_all(fd_pairs_[cur_pipe_idx].pipe_fds_.queen_fd_,(char*)&tsk, sizeof(void*)) != 0)
          {
            UL_LOG_ERROR("fail to write data to pipe [idx:%lld,errno:%d]", cur_pipe_idx, errno);
            queen_poll_fds_[cur_pipe_idx].fd = -1;
            close_pipe(fd_pairs_[cur_pipe_idx].fd_pair_.fd_);
          }
          else
          {
            next_push_pipe_idx_ = (cur_pipe_idx + 1)%consumer_num_;
            break;
          }
        }
        else
        {
          cur_pipe_idx ++;
          cur_pipe_idx = cur_pipe_idx%consumer_num_;
        }
      }
      if (checked_fd_num >= consumer_num_)
      {
        UL_LOG_ERROR("bad error, all pipes in the task queue was broken");
        errno = EPIPE;
        err = ERRNO2ERR(errno);
      }
      else
      {
        /// do nothing, find one correct pipe
      }
    }
    else
    {
      /// do nothing, find one idle pipe;
    }
  }
  return err;
}

int LinuxNetworkUtility::UlPipeQueenWorkerTQ::claim_task(void **tsk, const int64_t ntsk,  
                                                         const int64_t timeout_us, 
                                                         const int64_t worker_id)
{
  int err = 0;
  int32_t readed_len = 0;
  if (NULL == tsk || ntsk <= 0)
  {
    UL_LOG_WARNING("argument error [tsk:%p,ntsk:%lld]", tsk, ntsk);
    errno = EINVAL;
    err = ERRNO2ERR(errno);
  }
  if (0 == err && (worker_id < 0 || worker_id >= consumer_num_))
  {
    UL_LOG_WARNING("argument error, consumer id out of range [worker_i:%lld,consumer_num:%lld]", 
                   worker_id, consumer_num_);
    errno = EINVAL;
    err = ERRNO2ERR(errno);
  }
  if (0 == err && NULL == fd_pairs_)
  {
    errno = EINVAL;
    err = ERRNO2ERR(errno);
    UL_LOG_WARNING("task queue not initialized yet, initialize it first");
  }
  if (0 == err 
      &&  (fd_pairs_[worker_id].pipe_fds_.worker_fd_ < 0 
           || fd_pairs_[worker_id].pipe_fds_.queen_fd_   < 0)
     )
  {
    errno  = EPIPE;
    err = ERRNO2ERR(errno);
  }
  if (0 == err)
  {
    readed_len = fast_read(fd_pairs_[worker_id].pipe_fds_.worker_fd_, tsk, sizeof(void*)*ntsk, timeout_us);
    if (readed_len > 0 && readed_len % sizeof(void*) == 0)
    {
      /// do nothing
    }
    else
    {
      if (EAGAIN == errno || EINTR == errno)
      {
        err = ERRNO2ERR(errno);
      }
      else
      {
        queen_poll_fds_[worker_id].fd = -1;
        close_pipe(fd_pairs_[worker_id].fd_pair_.fd_);
        errno = EPIPE;
        err = ERRNO2ERR(errno);
      }
    }
  }
  if (0 == err && readed_len > 0)
  {
    err = readed_len/sizeof(void*);
  }
  return err;
}

int LinuxNetworkUtility::UlPipeQueenWorkerTQ::submit_output(void *tsk, const int64_t worker_id)
{
  int err = 0;
  if (NULL == tsk)
  {
    UL_LOG_WARNING("argument error [tsk:%p]", tsk);
    errno = EINVAL;
    err = ERRNO2ERR(errno);
  }
  if (0 == err && (worker_id < 0 || worker_id >= consumer_num_))
  {
    UL_LOG_WARNING("argument error, consumer id out of range [worker_i:%lld,consumer_num:%lld]", 
                   worker_id, consumer_num_);
    errno = EINVAL;
    err = ERRNO2ERR(errno);
  }
  if (0 == err)
  {
    if (0 > fd_pairs_[worker_id].fd_pair_.fd_[0] || 0 > fd_pairs_[worker_id].fd_pair_.fd_[0])
    {
      errno = EPIPE;
      err = ERRNO2ERR(errno);
      UL_LOG_WARNING("pipe broken");
    }
    else
    {
      if (write_all(fd_pairs_[worker_id].pipe_fds_.worker_fd_,(char*)&tsk, sizeof(void*)) != 0)
      {
        UL_LOG_WARNING("write to pipe fail [errno:%d]", errno);
        err = ERRNO2ERR(errno);
        queen_poll_fds_[worker_id].fd = -1;
        close_pipe(fd_pairs_[worker_id].fd_pair_.fd_);
      }
    }
  }
  return err;
}

int LinuxNetworkUtility::UlPipeQueenWorkerTQ::fetch_output(void **tsk, const int64_t ntsk, 
                                                           const int64_t timeout_us)
{
  int err = 0;
  int64_t poll_timeout_ms = 0;
  int64_t readable_fd_num = 0;
  UNUSED(readable_fd_num);
  int64_t got_tsk_num = 0;
  int32_t readed_len = 0;
  if (NULL == tsk || ntsk <= 0)
  {
    UL_LOG_WARNING("param error [tsk:%p,ntsk:%lld]", tsk, ntsk);
    errno = EINVAL;
    err = ERRNO2ERR(errno);
  }
  if (timeout_us <= 0)
  {
    poll_timeout_ms = 0;
  }
  else
  {
    poll_timeout_ms = (timeout_us + 999)/1000;
  }
  /// if (0 == err)
  /// {
  ///   for (int32_t i = 0; i < consumer_num_ &&  got_tsk_num < ntsk; i++)
  ///   {
  ///     if (queen_poll_fds_[i].fd > 0)
  ///     {
  ///       readed_len = read(fd_pairs_[i].pipe_fds_.queen_fd_, tsk + got_tsk_num,
  ///                         sizeof(void*)*(ntsk - got_tsk_num));
  ///       if (readed_len < 0 || readed_len%sizeof(void*) != 0)
  ///       {
  ///         if (readed_len < 0 &&( EAGAIN == errno || EINTR == errno ))
  ///         {
  ///           /// do nothing
  ///         }
  ///         else
  ///         {
  ///           UL_LOG_ERROR("fail to read data from pipe [idx:%lld,errno:%d]", i, errno);
  ///           queen_poll_fds_[i].fd = -1;
  ///           close_pipe(fd_pairs_[i].fd_pair_.fd_);
  ///         }
  ///       }
  ///       else
  ///       {
  ///         got_tsk_num  += readed_len/sizeof(void*);
  ///       }
  ///     }
  ///   }
  /// }
  if (0 == err)
  {
    readable_fd_num = poll(queen_poll_fds_, consumer_num_,poll_timeout_ms);
    if (0 == readable_fd_num)
    {
      errno = EAGAIN;
      err = ERRNO2ERR(errno);
    }
    else if (readable_fd_num < 0)
    {
      UL_LOG_ERROR("fail to poll queen fds, [errno:%d]", errno);
      clear();
      errno = EPIPE;
      err = ERRNO2ERR(errno);
    }
    else
    {
      for (int32_t i = 0; i < consumer_num_ &&  got_tsk_num < ntsk; i++)
      {
        if (queen_poll_fds_[i].revents & POLLIN)
        {
          readed_len = read(fd_pairs_[i].pipe_fds_.queen_fd_, tsk + got_tsk_num,
                            sizeof(void*)*(ntsk - got_tsk_num));
          if (readed_len < 0 || readed_len%sizeof(void*) != 0)
          {
            UL_LOG_ERROR("fail to read data from pipe [idx:%lld,errno:%d]", i, errno);
            queen_poll_fds_[i].fd = -1;
            close_pipe(fd_pairs_[i].fd_pair_.fd_);
          }
          else
          {
            got_tsk_num  += readed_len/sizeof(void*);
          }
        }
      }
    }
  }
  if (0 == err)
  {
    if (got_tsk_num > 0)
    {
      err = got_tsk_num;
    }
    else
    {
      errno = EAGAIN;
      err = ERRNO2ERR(errno);
    }
  }
  return err;
}
